package com.shux.kafka.consumer;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 描述：
 * 
 * <pre>HISTORY
 * ****************************************************************************
 *  ID   DATE           PERSON          REASON
 *  1    2017年9月19日      Simba.Hua         Create
 * ****************************************************************************
 * </pre>
 * @author Simba.Hua
 */
public class KafkaBatchMessage<T> {
	private List<KafkaMessage<T>> batchMessageList = new ArrayList<>();
	private Map<Integer,Long> offsets = new HashMap<>();//记录每个partition消费的最新的offset
	public void addKafkaMessage(KafkaMessage<T> message) {
		batchMessageList.add(message);
		offsets.put(message.getPartition(), message.getOffset());
	}
	public int getSize(){
		return batchMessageList.size();
	}
	public boolean isEmpty(){
		return batchMessageList.isEmpty();
	}
	public List<T> getMessages(){
		List<T> messages = new ArrayList<>();
		batchMessageList.forEach(m -> {
			messages.add(m.getMessage());
		});
		return messages;
	}
	public Map<Integer, Long> getOffsets() {
		return offsets;
	}
	public void removeKafkaMessage(int partition) {
		for(KafkaMessage<T> message : batchMessageList){
			if(message.getPartition() == partition) {
				batchMessageList.remove(message);
			}
		}
		offsets.remove(partition);
	}
}
